package com.weimob.mcloud.wechatwork.customer.mq.handler.externaluser;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
import com.google.common.collect.Lists;
import com.weimob.arch.artemis.client.api.consume.ArtemisMessage;
import com.weimob.arch.artemis.client.api.consume.ArtemisMessages;
import com.weimob.arch.artemis.client.spring.annotation.ArtemisMessageListener;
import com.weimob.arch.artemis.client.spring.annotation.Subscribe;
import com.weimob.arch.sub.env.application.annotation.SubEnvExposePoint;
import com.weimob.arch.sub.env.application.annotation.SubEnvTagExtractor;
import com.weimob.dq.client.annotation.DqListener;
import com.weimob.dq.core.entity.MsgPoolEntity;
import com.weimob.mcloud.mcloudratelimitercore.common.RatelimiterInfo;
import com.weimob.mcloud.mcloudratelimitercore.common.enums.BizLimitKeyEnum;
import com.weimob.mcloud.mcloudratelimitercore.common.enums.LimitKeyEnum;
import com.weimob.mcloud.mcloudratelimitercore.common.utils.LimitKeyUtils;
import com.weimob.mcloud.wechatwork.common.enums.CommonBizErrorEnum;
import com.weimob.mcloud.wechatwork.common.enums.MainBizDomainEnum;
import com.weimob.mcloud.wechatwork.common.enums.SubBizDomainEnum;
import com.weimob.mcloud.wechatwork.common.utils.BizErrorCodeGenerator;
import com.weimob.mcloud.wechatwork.core.adapter.pojo.dto.channel.base.RateLimit4CustomerDTO;
import com.weimob.mcloud.wechatwork.core.adapter.pojo.dto.channel.response.ExternalContactBO;
import com.weimob.mcloud.wechatwork.core.adapter.pojo.dto.channel.response.FollowUserBO;
import com.weimob.mcloud.wechatwork.core.adapter.pojo.dto.channel.response.SelfExternalContactGetBO;
import com.weimob.mcloud.wechatwork.core.adapter.service.ChannelAdapterService;
import com.weimob.mcloud.wechatwork.core.entity.privatesphere.PrivateSphere;
import com.weimob.mcloud.wechatwork.core.entity.relation.CustomerEmployeeRelation;
import com.weimob.mcloud.wechatwork.core.entity.relation.customer.Customer;
import com.weimob.mcloud.wechatwork.core.entity.relation.employee.Employee;
import com.weimob.mcloud.wechatwork.core.enums.customer.CustomerStatusEnum;
import com.weimob.mcloud.wechatwork.core.enums.customer.ExternalUserAddWayEnum;
import com.weimob.mcloud.wechatwork.core.enums.employee.EmployeeActiveStatusEnum;
import com.weimob.mcloud.wechatwork.core.enums.employee.EmployeeStatusEnum;
import com.weimob.mcloud.wechatwork.core.service.customer.CustomerService;
import com.weimob.mcloud.wechatwork.core.service.employee.EmployeeService;
import com.weimob.mcloud.wechatwork.core.service.privatesphere.PrivateSphereService;
import com.weimob.mcloud.wechatwork.customer.config.DqConfig;
import com.weimob.mcloud.wechatwork.customer.convertor.customer.CustomerConvert;
import com.weimob.mcloud.wechatwork.customer.domain.ArtemisBaseInfo;
import com.weimob.mcloud.wechatwork.customer.domain.ArtemisConstant;
import com.weimob.mcloud.wechatwork.customer.domain.ExternalUserChangeInfo;
import com.weimob.mcloud.wechatwork.customer.externalDependencyProxy.usermanagement.EmployeeProxy;
import com.weimob.mcloud.wechatwork.customer.mq.dto.AddExternalUserSecondDTO;
import com.weimob.mcloud.wechatwork.customer.mq.handler.externaluser.retry.AddExternalUserRetry;
import com.weimob.mcloud.wechatwork.customer.mq.model.ChatGroupFacade;
import com.weimob.mcloud.wechatwork.customer.service.CustomerTagSyncService;
import com.weimob.mcloud.wechatwork.customer.service.common.CustomerCommonService;
import com.weimob.mcloud.wechatwork.customer.service.impl.ArtemisService;
import com.weimob.mcloud.wechatwork.customer.service.user.base.BasicInfo;
import com.weimob.mcloud.wechatwork.customer.service.user.constant.RedisKeyConstant;
import com.weimob.mcloud.wechatwork.customer.service.user.enums.CustomerLoseTypeEnum;
import com.weimob.mcloud.wechatwork.customer.service.user.enums.YesOrNoEnum;
import com.weimob.mcloud.wechatwork.customer.third.ChannelAdapterServiceProxy;
import com.weimob.mcloud.wechatwork.customer.third.PrivateSphereThird;
import com.weimob.mcloud.wechatwork.customer.third.core.CustomerServiceProxy;
import com.weimob.mcloud.wechatwork.customer.third.customer.CustomerThird;
import com.weimob.soa.common.response.SoaResponse;
import com.weimob.soa.common.utils.SoaUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Create Time: 2021年01月18日 16:11
 *
 * @author maosheng.li
 **/
@Slf4j
@Service
public class AddCorpExternaluserHandler implements CorpExternalUserHandler {

    @Autowired
    private CustomerServiceProxy customerServiceProxy;

    @Reference
    private EmployeeService employeeService;

    @Autowired
    private ChannelAdapterService channelAdapterService;

    @Autowired
    private ArtemisService artemisService;

    @Reference
    private PrivateSphereService privateSphereService;

    @Autowired
    private PrivateSphereThird privateSphereThird;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private CustomerTagSyncService customerTagSyncService;

    @Autowired
    private CustomerCommonService customerCommonService;

    /**
     * 企微助手产品id
     */
    @Value("${aewc.productId:203}")
    private Long aewcProductId;

    @Autowired
    private EmployeeProxy employeeProxy;

    /**
     * widRetryFlag重试标识
     */
    @ApolloJsonValue("${wid.retry.flag.corpid:[]}")
    private List<String> widRetryCorpIds;

    @Autowired
    private CustomerThird customerThird;

    @Autowired
    AddExternalUserRetry addExternalUserRetry;

    @Resource
    RedisTemplate<String, RatelimiterInfo> redisTemplate;

    @Resource(name = "customerRelationChangeProcessThreadPool")
    ThreadPoolTaskExecutor customerRelationChangeProcessThreadPool;

    @Autowired
    ChannelAdapterServiceProxy channelAdapterServiceProxy;

    @Resource
    ChatGroupFacade chatGroupFacade;

    @Reference
    CustomerService customerService;

    @Override
    public SoaResponse<?, ?> handleCorpExternalUser(ExternalUserChangeInfo externalUserChageInfo) {
        log.info("添加外部联系人，start, AddCorpExternaluserHandler, param:{}", JSON.toJSONString(externalUserChageInfo));
        String corpId = externalUserChageInfo.getToUserName();
        String externalUserId = externalUserChageInfo.getExternalUserID();
        String userId = externalUserChageInfo.getUserID();
        String templateId = externalUserChageInfo.getTemplateId();

        PrivateSphere privateSphereRequest = new PrivateSphere();
        privateSphereRequest.setCorpId(corpId);
        privateSphereRequest.setProductId(aewcProductId);
        privateSphereRequest.setTemplateId(templateId);
        SoaResponse<PrivateSphere, Void> fetchPrivateSphere = privateSphereService.fetch(privateSphereRequest);
        if (0L != Long.parseLong(fetchPrivateSphere.getReturnCode()) || Objects.isNull(fetchPrivateSphere.getResponseVo())) {
            //1.1、私域不存在，返回（不落非企微产品外部联系人数据）
            log.warn("消费新增外部联系人消息查询私域信息失败,corpId:{}, templateId:{}", corpId, templateId);
            String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                    , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.ABNORMAL_DATA);
            return SoaUtil.error(errorCode, "消费新增外部联系人消息查询私域信息失败", null);

        }
        PrivateSphere privateSphere = fetchPrivateSphere.getResponseVo();
        Long bosId = privateSphere.getBosId();

        //消息加锁，最多等待0秒，上锁以后3秒自动解锁  针对同一员工和客户并发，重复的直接放过
        String handlerUk = String.format(RedisKeyConstant.ADD_CORP_USER_EXTERNAL_USER_LOCK, bosId, corpId, externalUserId, userId);
        try {
            RLock lock = redissonClient.getLock(handlerUk);
            boolean lockResult = lock.tryLock(0, 3, TimeUnit.SECONDS);
            if (!lockResult) {
                log.warn("消费外部联系人分布式锁过滤的消息，msgUniKey:{}", handlerUk);
                String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                        , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.ILLEGAL_ARGUMENTS);
                return SoaUtil.error(errorCode, "消息被消费外部联系人分布式锁过滤，msgUniKey：" + handlerUk, null);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("添加外部联系人消息，InterruptedException error", e);
        }

        String msgUk = String.format(RedisKeyConstant.ADD_CORP_USER_EXTERNAL_USER_CREATE_TIME, bosId, corpId, externalUserId, userId);
        //记录事件事件时间
        customerCommonService.recordEventTime(msgUk, externalUserChageInfo.getCreateTime());
        //2、判断员工是否存在 且在可见范围
        Employee employee = employeeProxy.fetchEmployeeById(privateSphere, userId, null);
        if (Objects.isNull(employee) || Objects.isNull(employee.getStatus())) {
            log.warn("handleCorpExternalUser.employee is null or status is null,userId:{}", userId);
            String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                    , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.ILLEGAL_ARGUMENTS);
            return SoaUtil.error(errorCode, "消费新增外部联系人消息查询员工为空", null);
        }
        // 默认有效 员工未授权或不在坐席则无效落数据
        CustomerStatusEnum customerStatus = CustomerStatusEnum.DEFAULT;
        if (employee.getActiveStatus() != null && EmployeeActiveStatusEnum.INACTIVE.getCode() == employee.getActiveStatus().getCode()
                || employee.getStatus() != null && EmployeeStatusEnum.USER_UN_AUTHORIZATION.getCode() == employee.getStatus().getCode()) {
            customerStatus = CustomerStatusEnum.UNAUTHORIZED;
        }

        //3、查询企微客户详情（需要接入企微限频）
        RateLimit4CustomerDTO<ExternalUserChangeInfo> rateLimitDTO = new RateLimit4CustomerDTO<>();
        rateLimitDTO.setTaskId(externalUserChageInfo.getTaskId());
        rateLimitDTO.setBizLimitKey(BizLimitKeyEnum.NOW_ADD_EXTERNAL_CONTACT_NOW.getKey());
        rateLimitDTO.setLimitKey(LimitKeyEnum.SLEF_EXTERNAL_CONTACT_GET.getKey());
        SoaResponse<SelfExternalContactGetBO, Void> selfExternalContactSoaResponse = channelAdapterServiceProxy.getSelfExternalContact(privateSphere, externalUserId, rateLimitDTO);
        String returnCode = selfExternalContactSoaResponse.getReturnCode();
        //3.1 接入限频
        if (CommonBizErrorEnum.WECHAT_WORK_LIMIT.getCode().equals(returnCode) || CommonBizErrorEnum.WECHAT_WORK_CONVERT_ID_LIMIT.getCode().equals(returnCode)) {
            SelfExternalContactGetBO responseVo = selfExternalContactSoaResponse.getResponseVo();
            if (Objects.nonNull(responseVo)) {
                Long taskId = responseVo.getTaskId();
                log.warn("添加外部联系人调用腾讯接口被限频，taskId:{},msg：{}", taskId, selfExternalContactSoaResponse.getReturnMsg());
                //创建一个限频的redis任务记录
                RatelimiterInfo ratelimiterInfo = new RatelimiterInfo(rateLimitDTO.getLimitKey(), JSON.toJSONString(externalUserChageInfo), rateLimitDTO.getBizLimitKey());
                redisTemplate.opsForValue().set(LimitKeyUtils.buildRedisKey(taskId), ratelimiterInfo, 1, TimeUnit.DAYS);

              /*  String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                        , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.WECHAT_WORK_LIMIT);*/
                return SoaUtil.error(returnCode, "添加外部联系人调用腾讯接口被限频，taskId:" + taskId, null);
            }
        }
        if (0L != Long.parseLong(returnCode) || Objects.isNull(selfExternalContactSoaResponse.getResponseVo())) {
            log.warn("实时-添加客户消息, 查询腾讯客户详情信息失败, response:{}", JSON.toJSONString(selfExternalContactSoaResponse));
            String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                    , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.ABNORMAL_DATA);
            return SoaUtil.error(errorCode, "实时-添加客户消息, 查询腾讯客户详情信息失败", null);
        }

        try {
            SelfExternalContactGetBO selfExternalContactGetBO = selfExternalContactSoaResponse.getResponseVo();
            ExternalContactBO externalContactBO = selfExternalContactGetBO.getExternalContactBO();
            List<FollowUserBO> followUserList = selfExternalContactGetBO.getFollowUser();
            String phone = customerCommonService.getCustomerPhone(followUserList);

            //5、查询本地企微用户信息（没有需要写入）
            List<Customer> customerResponseList = null;
            try {
                customerResponseList = customerServiceProxy.fetchCustomerByIds(null, externalUserId, privateSphere);
            } catch (Exception e) {
                log.error("查询企微助手本地客户基本信息失败，error", e);
            }

            Boolean newUserFlag = false;
            Customer customer = CustomerConvert.toCustomer(privateSphere, externalUserId, phone, externalContactBO);

            if (CollectionUtils.isEmpty(customerResponseList)) {
                //5.1 如果本地不存在，新增外部联系人数据 客户状态为此时客户员工关系的状态
                customer.setStatus(customerStatus);
                // 首次添加事件，只有全新客户存一下
                Date firstAddTime = customerCommonService.getCustomerFirstAddTime(followUserList);
                customer.setFirstAddTime(firstAddTime);
                SoaResponse<Customer, ?> customerSoaResponse = customerService.createCustomer(customer);
                if (0L != Long.parseLong(customerSoaResponse.getReturnCode())) {
                    log.error("实时好友消息-创建本地客户数据失败,customerSoaResponse:{}", JSON.toJSONString(customerSoaResponse));
                    String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                            , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.INTERNAL_ERROR);
                    return SoaUtil.error(errorCode, "实时好友消息-创建本地客户数据失败", null);
                }
                customer = customerSoaResponse.getResponseVo();
                newUserFlag = true;
            } else {
                //5.2 如果本地数据存在，更新外部联系人最新数据 有效的不要变为无效 还有别的有效关系
                customer.setId(customerResponseList.get(0).getId());
                if (CustomerStatusEnum.UNAUTHORIZED.getCode().equals(customerResponseList.get(0).getStatus().getCode())) {
                    customer.setStatus(customerStatus);
                } else {
                    customer.setStatus(customerResponseList.get(0).getStatus());
                }

                SoaResponse<Customer, ?> customerSoaResponse = customerService.updateCustomer(customer);
                if (0L != Long.parseLong(customerSoaResponse.getReturnCode())) {
                    log.error("更新企微助手本地客户基本信息失败, response:{}", JSON.toJSONString(customerSoaResponse));

                    String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                            , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.INTERNAL_ERROR);
                    return SoaUtil.error(errorCode, "更新企微助手本地客户基本信息失败", null);
                }

                customer = customerSoaResponse.getResponseVo();
                if (phone != null) {
                    SoaResponse<Customer, ?> customerDetailSoaResponse = customerService.updateCustomerDeatil(customer);
                    if (0L != Long.parseLong(customerDetailSoaResponse.getReturnCode())) {
                        log.error("更新企微助手本地客户详情信息失败, request:{}, response:{}",
                                JSON.toJSONString(customer), JSON.toJSONString(customerDetailSoaResponse));
                    }
                }
            }

            //6、查询客户员工关系数据（如果不存在需要写入，存在需要更新）查询该客户所有关联员工是为了给二阶段业务处理使用
            List<CustomerEmployeeRelation> relationList;
            try {
                relationList = customerServiceProxy.fetchCustomerEmployeeRelation(customer.getId(), null, privateSphere, null, null, null);
            } catch (Exception e) {
                log.error("查询企微助手本地客户员工关系失败, customerId:{},error", customer.getId(), e);
                String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                        , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.INTERNAL_ERROR);
                return SoaUtil.error(errorCode, "查询企微助手本地客户员工关系失败", null);
            }

            Map<String, CustomerEmployeeRelation> customerEmployeeRelationMap = new LinkedHashMap<>();
            if (CollectionUtils.isNotEmpty(relationList)) {
                for (CustomerEmployeeRelation employeeRelation : relationList) {
                    Employee emp = employeeRelation.getEmployee();
                    if (Objects.nonNull(emp)) {
                        customerEmployeeRelationMap.put(emp.getId(), employeeRelation);
                    }
                }
            }


            //实时消息，只用处理当前客户加的员工
            List<FollowUserBO> filterFollowUserBo = followUserList.stream().filter(item -> item.getUserid().equals(userId)).collect(Collectors.toList());
            if (CollectionUtils.isEmpty(filterFollowUserBo)) {
                log.warn("添加外部联系人好友回调，filterFollowUserBo is empty userId:{}", userId);
                String errorCode = BizErrorCodeGenerator.generate(MainBizDomainEnum.CUSTOMER_MANAGEMENT
                        , SubBizDomainEnum.CUSTOMER_LIST, CommonBizErrorEnum.ABNORMAL_DATA);
                return SoaUtil.error(errorCode, "添加外部联系人好友回调，filterFollowUserBo is empty userId:" + userId, null);
            }

            String customerId = customer.getId();
            String employeeId = employee.getId();
            Boolean userStateChangeFlag = false;
            for (FollowUserBO followUserBO : filterFollowUserBo) {
                //6.1 convert
                Date followUserCreateTime = new Date(followUserBO.getCreatetime() * 1000L);
                CustomerEmployeeRelation customerEmployeeRelation = CustomerConvert.
                        toCustomerEmployeeRelation(corpId, customerId, employee, followUserCreateTime, followUserBO);

                CustomerEmployeeRelation customerEmployeeRelationResult = customerEmployeeRelationMap.get(employeeId);
                if (Objects.isNull(customerEmployeeRelationResult)) {
                    //6.1.1 本地不存在该员工客户关系，新增数据
                    log.info("新中台消费外部联系人新增客户员工关系： {}", JSON.toJSONString(customerEmployeeRelation));
                    customerEmployeeRelation.setFriended(true);
                    customerService.createCustomerEmployeeRelation(customerEmployeeRelation);
                    userStateChangeFlag = true;
                } else {
                    if (YesOrNoEnum.YES.getCode().equals(externalUserChageInfo.getHistoryCustomerSync())) {
                        //6.1.4 历史同步事件
                        if (CustomerLoseTypeEnum.EMPLOYEE_DEL_CUSTOMER.getCode().equals(customerEmployeeRelationResult.getLoseType())) {
                            //6.1.5 单向删除员工删除客户，需要将流失类型置为初识标识
                            customerEmployeeRelation.setLoseType(CustomerLoseTypeEnum.INIT.getCode());
                            customerEmployeeRelation.setFriended(Boolean.TRUE);
                        }
                    } else {
                        customerEmployeeRelation.setLoseType(CustomerLoseTypeEnum.INIT.getCode());
                        customerEmployeeRelation.setFriended(Boolean.TRUE);
                    }

                    String localState = customerEmployeeRelationResult.getState();
                    Date localAddTime = customerEmployeeRelationResult.getAddTime();
                    String txState = followUserBO.getState();
                    //数据同步状态本地和企微state不一致需要通知裂变活动
                    if (!Objects.equals(localState, txState) && txState != null) {
                        userStateChangeFlag = true;
                    }
                    //数据同步状态本地和企微添加好友时间不一致需要通知裂变活动
                    if (followUserCreateTime.compareTo(localAddTime) != 0) {
                        userStateChangeFlag = true;
                    }

                    Integer loseType = customerEmployeeRelationResult.getLoseType();
                    //存在客户删除员工，腾讯查询详情接口仍然查得到好友关系，但产品上认为是非好友，所以在重试时需要判断这种场景,重试+客户删除员工时不需要更新，其余场景需要更新
                    if (!(CustomerLoseTypeEnum.CUSTOMER_DEL_EMPLOYEE.getCode().equals(loseType) && Boolean.TRUE.equals(externalUserChageInfo.getRetryFlag()))) {
                        customerService.updateCustomerEmployeeRelation(customerEmployeeRelation);
                    }
                }

                try {
                    //调用新的客户标签同步方法
                    customerTagSyncService.syncCustomerTag(privateSphere, externalUserId, null, followUserList, employee);
                } catch (Exception e) {
                    log.error("新增客户, 添加客户标签失败, corpId:{}, externalUserId:{}, error:",
                            corpId, externalUserId, e);
                }

                //6.1.8 发消息进行二阶段的操作(注意消息体不能过大，过滤无效字段数据)
                AddExternalUserSecondDTO addExternalUserSecondDTO = new AddExternalUserSecondDTO();
                addExternalUserSecondDTO.setCustomerId(customer.getId());
                addExternalUserSecondDTO.setBosId(bosId);
                addExternalUserSecondDTO.setNewUserFlag(newUserFlag);
                addExternalUserSecondDTO.setRemarkMobiles(followUserBO.getRemarkMobiles());
                addExternalUserSecondDTO.setCurrentRelation(filterInvalidRelationAtt(customerEmployeeRelation));
                addExternalUserSecondDTO.setUserStateChangeFlag(userStateChangeFlag);
                if (MapUtils.isNotEmpty(customerEmployeeRelationMap)) {
                    List<CustomerEmployeeRelation> results = Lists.newArrayList();
                    for (CustomerEmployeeRelation value : customerEmployeeRelationMap.values()) {
                        CustomerEmployeeRelation filterResult = filterInvalidRelationAtt(value);
                        results.add(filterResult);
                    }
                    addExternalUserSecondDTO.setHisRelationList(results);
                }
                addExternalUserSecondDTO.setExternalUserChangeInfo(externalUserChageInfo);
                ArtemisBaseInfo<AddExternalUserSecondDTO> artemisBaseInfo = new ArtemisBaseInfo();
                //实时消息
                artemisBaseInfo.setTopic(ArtemisConstant.Topic.CORP_EXTERNALUSER_SECOND_EVENT);
                artemisBaseInfo.setEvent(addExternalUserSecondDTO);
                artemisBaseInfo.setProducer(ArtemisConstant.USER);
                artemisService.sendToArtemis(artemisBaseInfo, bosId);
            }

            //5、客户群好友关系维护
            Customer finalCustomer = customer;
            customerRelationChangeProcessThreadPool.execute(() -> {
                chatGroupFacade.execute(privateSphere, finalCustomer.getExternalUserid(), finalCustomer.getId(), false);
            });

        } catch (Exception e) {
            log.error("添加好友事件处理失败，AddCorpExternalUserHandler error", e);
        }
        return SoaUtil.ok(null);
    }


    /**
     * 客户员工关系转换去除无效数据
     *
     * @param customerEmployeeRelation
     * @return
     */
    private CustomerEmployeeRelation filterInvalidRelationAtt(CustomerEmployeeRelation customerEmployeeRelation) {
        CustomerEmployeeRelation customerEmployeeRelationResult = new CustomerEmployeeRelation();
        Employee employeeInfo = customerEmployeeRelation.getEmployee();
        if (employeeInfo != null) {
            Employee employeeResult = new Employee();
            employeeResult.setId(employeeInfo.getId());
            employeeResult.setName(employeeInfo.getName());
            employeeResult.setTxUserId(employeeInfo.getTxUserId());
            employeeResult.setVidList(employeeInfo.getVidList());
            employeeResult.setHigherVidList(employeeInfo.getHigherVidList());
            employeeResult.setActionLogs(null);
            employeeResult.setCustomers(null);
            employeeResult.setDepartments(null);
            employeeResult.setExtInfo(null);
            employeeResult.setTagDetails(null);
            employeeResult.setGroups(null);
            customerEmployeeRelationResult.setEmployee(employeeResult);
        }
        customerEmployeeRelationResult.setAddTime(customerEmployeeRelation.getAddTime());
        customerEmployeeRelationResult.setState(customerEmployeeRelation.getState());
        customerEmployeeRelationResult.setFriended(customerEmployeeRelation.getFriended());
        ExternalUserAddWayEnum addWayEnum = customerEmployeeRelation.getAddWay();
        if (Objects.isNull(addWayEnum)) {
            customerEmployeeRelationResult.setAddWay(ExternalUserAddWayEnum.UNKNOWN);
        } else {
            customerEmployeeRelationResult.setAddWay(addWayEnum);
        }
        customerEmployeeRelationResult.setSubAddWay(customerEmployeeRelation.getSubAddWay());
        customerEmployeeRelationResult.setStatus(customerEmployeeRelation.getStatus());
        customerEmployeeRelationResult.setExtInfo(null);
        return customerEmployeeRelationResult;
    }

    /**
     * 延迟队列的间接灰度方案使用，当延迟队列支持灰度后，可直接消费延迟队列
     *
     * 背景：延迟队列不支持灰度，Artemis支持灰度
     * 临时灰度实现方案：将接收到的延迟队列消息打上灰度标，再转发到Artemis，以此实现延迟队列的间接灰度
     *
     * 为减少理解成本和风险，需要做到
     * 1、延迟队列的topic名称和Artemis的topic名称保持一致
     * 2、延迟队列的消息体和Artemis的消息体保持一致
     * 3、延迟队列处理方法的入参和逻辑不改动，方便在延迟队列支持灰度后直接切换
     */
    @DqListener(topic = DqConfig.CUSTOMER_SEARCH_MSG_TOPIC)
    public void handlerCustomerSearchHacker(MsgPoolEntity msgPoolEntity) {
        if (null == msgPoolEntity) {
            return;
        }

        ExternalUserChangeInfo externalUserChangeInfo = this.getPushDqQueueModel(msgPoolEntity);
        if (null == externalUserChangeInfo) {
            log.error("ExternalUserChangeInfo is null. MsgPoolEntity:{}", JSON.toJSONString(msgPoolEntity));
            return;
        }

        if (null != externalUserChangeInfo.getBosId()) {
            this.push(externalUserChangeInfo.getBosId(), msgPoolEntity);
        } else {
            log.error("ExternalUserChangeInfo is illegal. ExternalUserChangeInfo:{}", JSON.toJSONString(externalUserChangeInfo));
        }
    }

    /**
     * 将延迟队列消息打上灰度标，原样推送至Artemis
     *
     * 延迟队列的间接灰度方案使用，当延迟队列支持灰度后，可直接消费延迟队列
     */
    @SubEnvExposePoint
    @SubEnvTagExtractor(key = "weimob-bosid", value = "#args[0]")
    public void push(long bosId, MsgPoolEntity msgPoolEntity) {
        ArtemisBaseInfo<MsgPoolEntity> artemisBaseInfo = new ArtemisBaseInfo<>();
        artemisBaseInfo.setTopic(DqConfig.CUSTOMER_SEARCH_MSG_TOPIC);
        artemisBaseInfo.setProducer(ArtemisConstant.USER);
        artemisBaseInfo.setEvent(msgPoolEntity);

        boolean success = this.artemisService.sendToArtemis(artemisBaseInfo);
        if (success) {
            log.info("ArtemisProducer send message success. bosId:{} ArtemisBaseInfo:{}", bosId, JSON.toJSONString(artemisBaseInfo));
        }
    }

    /**
     * 接收带灰度标的Artemis消息，并调用原延迟队列处理方法
     *
     * 延迟队列的间接灰度方案使用，当延迟队列支持灰度后，可直接消费延迟队列
     */
    @ArtemisMessageListener(value = @Subscribe(topic = DqConfig.CUSTOMER_SEARCH_MSG_TOPIC, user = ArtemisConstant.CONSUMER_USER))
    public void pull(ArtemisMessages artemisMessages) {
        if (CollectionUtils.isEmpty(artemisMessages)) {
            return;
        }

        for (ArtemisMessage artemisMessage : artemisMessages) {
            if (null == artemisMessage) {
                continue;
            }

            this.handlerCustomerSearch(JSON.parseObject(artemisMessage.getValue(), MsgPoolEntity.class));
        }
    }

    /**
     * wid未换取到，需要延迟重试
     *
     * @param msgClientEntity
     */
    public void handlerCustomerSearch(MsgPoolEntity msgClientEntity) {
        log.info("handlerCustomerSearch msg:{}", JSON.toJSONString(msgClientEntity));
        ExternalUserChangeInfo externalUserChangeInfo = this.getPushDqQueueModel(msgClientEntity);
        if (externalUserChangeInfo == null) {
            return;
        }
        String corpId = externalUserChangeInfo.getToUserName();
        String externalUserId = externalUserChangeInfo.getExternalUserID();
        String userId = externalUserChangeInfo.getUserID();
        Long bosId = externalUserChangeInfo.getBosId();
        RLock lock = redissonClient.getLock(String.format(
                RedisKeyConstant.CUSTOMER_SEARCH_RETRY, corpId, externalUserId, userId, bosId));

        try {
            //尝试加锁，最多等待0秒，上锁以后3秒自动解锁  针对同一员工和客户并发，重复的直接放过
            Boolean lockResult = lock.tryLock(0, 3, TimeUnit.SECONDS);

            if (!lockResult) {
                log.info("消费客户搜索重试幂等处理:{}", JSON.toJSONString(externalUserChangeInfo));
                return;
            }
            //先判断wid是否生成,未生成则再次丢到延迟队列
            BasicInfo basicInfo = new BasicInfo();
            basicInfo.setCorpId(corpId);
            basicInfo.setBosId(bosId);
            PrivateSphere privateSphere = privateSphereThird.fetch(basicInfo, Boolean.TRUE);
            Long superWid = customerThird.getSuperWid(privateSphere.getBosId(), externalUserId, corpId);
            if (Objects.isNull(superWid) || widRetryCorpIds.contains(corpId)) {
                log.warn("延迟队列里仍然未获取到wid,externalUserId:{}", externalUserId);
                //执行重试
                addExternalUserRetry.addExternalUserRetry(externalUserChangeInfo, privateSphere);
                return;
            }
            //执行重试
            this.handleCorpExternalUser(externalUserChangeInfo);

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("handlerCustomerSearch InterruptedException", e);
        } finally {
            if (lock != null && lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }

    /**
     * 获取延迟队列消息并对象转换
     */
    private ExternalUserChangeInfo getPushDqQueueModel(MsgPoolEntity msgClientEntity) {
        String body = msgClientEntity.getBody();
        if (org.apache.commons.lang3.StringUtils.isBlank(body)) {
            return null;
        }
        ExternalUserChangeInfo externalUserChangeInfo;
        try {
            externalUserChangeInfo = JSON.parseObject(body, ExternalUserChangeInfo.class);
        } catch (Exception e) {
            log.error("handlerConsumer json parse fail,msgClientEntity:{},case by:{}", JSON.toJSONString(msgClientEntity), e.getMessage());
            return null;
        }
        return externalUserChangeInfo;
    }

}
